package com.study.spark.scala.udf

import org.apache.spark.sql.SparkSession

/**
 *
 * @author stephen
 * @date 2019-09-27 09:50
 */
object UDFDemo {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("UDF Demo")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val df=Range(0,10).toSeq.toDF("id")
    // 定义一个函数，对给定的整数列都加100
    def add100(value:Int):Int = { value + 100 }

    // 1、Spark SQL用法
    // 注册成自定义sql函数
    spark.udf.register("add100", add100 _)
    //spark.udf.register("add100", add100(_:Int))
    // 调用上面写的自定义函数add100(value:Int)
    df.selectExpr("id", "add100(id) as new_id").show
    // 或者
    df.createOrReplaceTempView("tmp")
    spark.sql("select id,add100(id) as new_id from tmp").show

    // 2、DataFrame用法
    // 用spark.udf.register注册的函数，不能用作dataset的函数使用
    // 直接用的话，会类型不匹配
    // df.select(add100($"id")).show

    // 正确用法，用udf注册
    val add100_func = udf(add100 _)
    // 现在相当于有了一个add100_func的函数，类型是UserDefinedFunction
    df.select($"id", add100_func($"id").as("new_id")).show
    // 或者
    df.withColumn("new_id",add100_func($"id")).show()
    // withColumn和select的区别
    // withColumn的功能是实现增加一列，或者替换一个已存在的列
    // select方法和sql里的select一样，如果新增的列名在表里已经存在，那么结果里允许出现两列列名相同但数据不一样


    spark.stop()
  }
}
